package sparkStreaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.junit.Test

class SocketSource {

	@Test
	def socketTest(): Unit = {
		val conf: SparkConf = new SparkConf().setAppName("wordcount").setMaster("local[*]")

		// 构建 StreamingContext
		val streamingContext = new StreamingContext(conf, Seconds(3))

		// 从指定的端口采集数据,
		val socketLineDStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("localhost", 9999)

		// 将采集的数据进行分解
		val wordDStream: DStream[String] = socketLineDStream.flatMap(_.split(" "))

		// 将结构转换便于统计
		val mapDStream: DStream[(String, Int)] = wordDStream.map((_, 1))

		// reduceByKey 聚合
		val sumDStream: DStream[(String, Int)] = mapDStream.reduceByKey(_+_)

		// 打印结果，在Driver打印
		sumDStream.print()

		// 不能停止采集程序
		//streamingContext.stop()

		// 启动接收器
		streamingContext.start()
		// Driver等待接收器停止
		streamingContext.awaitTermination()
	}

	/**
	 * 新建文件拖进去
	 */
	@Test
	def fileTest(): Unit = {
		val conf: SparkConf = new SparkConf().setAppName("wordcount").setMaster("local[*]")

		// 构建 StreamingContext
		val streamingContext = new StreamingContext(conf, Seconds(5))

		// 从指定的端口采集数据,
		val fileLineDStream: DStream[String] = streamingContext.textFileStream("in/test")

		// 将采集的数据进行分解
		val wordDStream: DStream[String] = fileLineDStream.flatMap(_.split(" "))

		// 将结构转换便于统计
		val mapDStream: DStream[(String, Int)] = wordDStream.map((_, 1))

		// reduceByKey 聚合
		val sumDStream: DStream[(String, Int)] = mapDStream.reduceByKey(_+_)

		// 打印结果，在Driver打印
		sumDStream.print()

		// 不能停止采集程序
		//streamingContext.stop()

		// 启动接收器
		streamingContext.start()
		// Driver等待接收器停止
		streamingContext.awaitTermination()
	}
}
